package com.atguigu.edu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.bean.TestBean;
import com.atguigu.edu.util.ClickHouseUtil;
import com.atguigu.edu.util.DateFormatUtil;
import com.atguigu.edu.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class DwsTestExam {
    public static void main(String[] args) throws Exception {
        // TODO 1 环境准备
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // TODO 2 设置状态后端

        // TODO 3 读取数据
        String testTopic = "dwd_exam_detail";
        String groupId = "dws_test_exam";
        DataStreamSource<String> testStream =
                env.addSource(KafkaUtil.getFlinkKafkaConsumer(testTopic,
                        groupId));

        // TODO 4 过滤转换数据结构
        SingleOutputStreamOperator<JSONObject> jsonObjStream =
                testStream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                String userId = jsonObject.getString("user_id");
                if (userId != null) {
                    out.collect(jsonObject);
                }
            }
        });

        // TODO 5 根据试卷id进行分组
        KeyedStream<JSONObject, String> keyedStream =
                jsonObjStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject value) throws Exception {
                return value.getString("paper_id");
            }
        });

        SingleOutputStreamOperator<TestBean> beanStream =
                keyedStream.map(new MapFunction<JSONObject, TestBean>() {
            @Override
            public TestBean map(JSONObject value) throws Exception {
                return TestBean.builder()
                        .testPersonCount(1L)
                        .avgScore(value.getDouble("score"))
                        .avgDuringTime(value.getDouble("duration_sec"))
                        .ts(value.getLong("ts") * 1000)
                        .build();
            }
        });

        // TODO 6 开窗聚合
        SingleOutputStreamOperator<TestBean> beanWithWaterMarkStream = beanStream.assignTimestampsAndWatermarks(WatermarkStrategy.<TestBean>forBoundedOutOfOrderness(Duration.ofSeconds(2L)).withTimestampAssigner(new SerializableTimestampAssigner<TestBean>() {
            @Override
            public long extractTimestamp(TestBean element,
                                         long recordTimestamp) {
                return element.getTs();
            }
        }));

        SingleOutputStreamOperator<TestBean> reduceStream =
                beanWithWaterMarkStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(10L)))
                .reduce(new ReduceFunction<TestBean>() {
                    @Override
                    public TestBean reduce(TestBean value1, TestBean value2) throws Exception {
                        value1.setTestPersonCount(value1.getTestPersonCount() + value2.getTestPersonCount());
                        value1.setAvgScore(value1.getAvgScore() + value2.getAvgScore());
                        value1.setAvgDuringTime(value1.getAvgDuringTime() + value2.getAvgDuringTime());
                        return value1;
                    }
                }, new AllWindowFunction<TestBean, TestBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow window,
                                      Iterable<TestBean> values,
                                      Collector<TestBean> out) throws Exception {
                        TestBean next = values.iterator().next();
                        next.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        next.setTs(System.currentTimeMillis());
                        out.collect(next);
                    }
                });

        // TODO 7 写出到clickhouse
        reduceStream.addSink(ClickHouseUtil.getClickHouseSink("insert into dws_test_exam values(?,?,?,?,?,?)"));

        // TODO 8 执行任务
        env.execute(groupId);

    }
}
